#pragma once

#include <memory>
#include <vector>
#include <algorithm>

#include <odb/transaction.hxx>
#include <odb/query.hxx>
#include <odb/result.hxx>

#include <boost/date_time/posix_time/posix_time.hpp>

#include "logger.hpp"
#include "mysql.hpp"
#include "mysql_ptime_traits.hpp"
#include "outbox_event.hxx"
#include "outbox_event-odb.hxx"

namespace suye
{
class OutboxTable
{
public:
    using ptr = std::shared_ptr<OutboxTable>;

    explicit OutboxTable(const std::shared_ptr<odb::core::database>& db)
        : _db(db)
    {
    }

    bool record_event(const std::shared_ptr<OutboxEvent>& event)
    {
        if (!event)
        {
            LOG_ERROR("记录 Outbox 事件失败：事件对象为空！");
            return false;
        }

        if (!odb::transaction::has_current())
        {
            LOG_ERROR("记录 Outbox 事件失败：当前线程没有激活的事务！");
            return false;
        }

        try
        {
            _db->persist(*event);
            return true;
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("记录 Outbox 事件失败：{} - {}！", event->event_id(), e.what());
            return false;
        }
    }

    std::shared_ptr<OutboxEvent> find_by_event_id(const std::string& event_id)
    {
        std::shared_ptr<OutboxEvent> res;
        try
        {
            odb::transaction trans(_db->begin());
            typedef odb::query<OutboxEvent> query;
            res.reset(_db->query_one<OutboxEvent>(query::event_id == event_id));
            trans.commit();
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("查询 Outbox 事件失败：{} - {}！", event_id, e.what());
        }
        return res;
    }

    bool truncate()
    {
        try
        {
            odb::transaction trans(_db->begin());
            _db->execute("TRUNCATE TABLE `outbox_event`");
            trans.commit();
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("清空 Outbox 事件表失败：{}！", e.what());
            return false;
        }
        return true;
    }

    std::vector<std::shared_ptr<OutboxEvent>> claim_batch(std::size_t limit,
                                                          const boost::posix_time::ptime& now = boost::posix_time::second_clock::universal_time())
    {
        std::vector<std::shared_ptr<OutboxEvent>> claimed;
        if (limit == 0)
        {
            return claimed;
        }

        try
        {
            odb::transaction trans(_db->begin());
            typedef odb::query<OutboxEvent> query;
            typedef odb::result<OutboxEvent> result;

            const unsigned char pending = static_cast<unsigned char>(OutboxStatus::Pending);
            const unsigned char failed = static_cast<unsigned char>(OutboxStatus::Failed);

            result r(_db->query<OutboxEvent>(
                (query::status == pending && query::available_at <= now) ||
                (query::status == failed && query::retry_count < query::max_retry && query::available_at <= now)));

            std::vector<std::shared_ptr<OutboxEvent>> candidates;
            for (auto it = r.begin(); it != r.end(); ++it)
            {
                candidates.emplace_back(std::make_shared<OutboxEvent>(*it));
            }

            std::sort(candidates.begin(),
                      candidates.end(),
                      [](const std::shared_ptr<OutboxEvent>& lhs, const std::shared_ptr<OutboxEvent>& rhs) {
                          return lhs->id() < rhs->id();
                      });

            for (auto& event_ptr : candidates)
            {
                if (claimed.size() >= limit)
                {
                    break;
                }
                event_ptr->status(OutboxStatus::Dispatching);
                event_ptr->available_at(now);
                event_ptr->updated_at(now);
                _db->update(*event_ptr);
                claimed.emplace_back(event_ptr);
            }

            trans.commit();
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("拉取 Outbox 事件失败：{}！", e.what());
        }

        return claimed;
    }

    bool mark_processed(const std::string& event_id)
    {
        try
        {
            odb::transaction trans(_db->begin());
            typedef odb::query<OutboxEvent> query;

            std::shared_ptr<OutboxEvent> event(_db->query_one<OutboxEvent>(query::event_id == event_id));
            if (!event)
            {
                LOG_WARN("标记 Outbox 事件成功失败：事件 {} 不存在！", event_id);
                return false;
            }

            const auto now = boost::posix_time::second_clock::universal_time();
            event->status(OutboxStatus::Completed);
            event->updated_at(now);
            event->available_at(now);
            event->last_error("");
            _db->update(*event);
            trans.commit();
            return true;
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("标记 Outbox 事件成功失败：{} - {}！", event_id, e.what());
        }
        return false;
    }

    bool mark_failed(const std::string& event_id,
                     const std::string& reason,
                     const boost::posix_time::time_duration& backoff = boost::posix_time::seconds(5))
    {
        try
        {
            odb::transaction trans(_db->begin());
            typedef odb::query<OutboxEvent> query;

            std::shared_ptr<OutboxEvent> event(_db->query_one<OutboxEvent>(query::event_id == event_id));
            if (!event)
            {
                LOG_WARN("标记 Outbox 事件失败状态失败：事件 {} 不存在！", event_id);
                return false;
            }

            const auto now = boost::posix_time::second_clock::universal_time();
            const unsigned int next_retry = event->retry_count() + 1;
            event->retry_count(next_retry);
            event->last_error(reason);
            event->updated_at(now);

            if (next_retry >= event->max_retry())
            {
                event->status(OutboxStatus::Failed);
                event->available_at(now);
            }
            else
            {
                event->status(OutboxStatus::Pending);
                event->available_at(now + backoff);
            }

            _db->update(*event);
            trans.commit();
            return true;
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("标记 Outbox 事件失败状态失败：{} - {}！", event_id, e.what());
        }
        return false;
    }

    std::size_t requeue_stuck_events(const boost::posix_time::time_duration& timeout,
                                     const boost::posix_time::ptime& now = boost::posix_time::second_clock::universal_time())
    {
        std::size_t requeued = 0;
        try
        {
            odb::transaction trans(_db->begin());
            typedef odb::query<OutboxEvent> query;
            typedef odb::result<OutboxEvent> result;

            const unsigned char dispatching = static_cast<unsigned char>(OutboxStatus::Dispatching);
            const boost::posix_time::ptime deadline = now - timeout;

            result r(_db->query<OutboxEvent>(query::status == dispatching && query::updated_at <= deadline));
            for (auto it = r.begin(); it != r.end(); ++it)
            {
                auto event_ptr = std::make_shared<OutboxEvent>(*it);
                event_ptr->status(OutboxStatus::Pending);
                event_ptr->available_at(now);
                event_ptr->updated_at(now);
                _db->update(*event_ptr);
                ++requeued;
            }

            trans.commit();
        }
        catch (const std::exception& e)
        {
            LOG_ERROR("重置卡顿的 Outbox 事件失败：{}！", e.what());
        }
        return requeued;
    }

private:
    std::shared_ptr<odb::core::database> _db;
};
}
